package chapter4.eg12;

import java.util.Map;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

public class TimestampPrependerInterceptor implements ProducerInterceptor<String, String> {

  @Override
  public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
    System.out.println("start interceptor");
    return new ProducerRecord<>(record.topic(), record.partition(), record.timestamp(),
      record.key(), System.currentTimeMillis() + "," + record.value().toString());
  }

  @Override
  public void onAcknowledgement(RecordMetadata metadata, Exception exception) {

  }

  @Override
  public void close() {

  }

  @Override
  public void configure(Map<String, ?> configs) {

  }
}
